package cn.com.guage.monitor.log.kafka.log;

import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.classic.spi.LoggingEvent;
import ch.qos.logback.core.rolling.RollingFileAppender;
import cn.com.yusys.yusp.commons.log.common.util.StringUtil;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static cn.com.yusys.yusp.commons.log.common.appender.FieldNameConst.*;

/**
 * json格式日志的appender
 * 给每个json日志加了公共字段：应用名称、时间戳、ip、端口号
 * 为了减小日志输出到文件的磁盘io，这里实现了日志的缓存、定时输出、批量输出
 *
 * @author xulifei
 * @since 2.3.1
 */
public class JsonAppender extends RollingFileAppender<ILoggingEvent> {
	
    private static Logger logger = (Logger) LoggerFactory.getLogger(JsonAppender.class);

    /**
     * 日志的生成时间格式：这里采用的是filebeat默认的时间格式
     */
    private static final String LOG_TIMESTAMP_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";

    /**
     * 日志输出频率（单位：秒）
     */
    private long freq = 10;

    /**
     * 缓存容量（单位：条）
     */
    private int cacheCapacity = 128;

    /**
     * 忽略添加的公共字段，多个用逗号分开
     */
    private String ignoreCommonFields;

    /**
     * true表示禁止添加所有公共字段
     */
    private boolean noCommonFields = false;

    /**
     * 公共字段拼接后的字符串(值不变化的字段)
     */
    private String finalFieldsStr = "";

    /**
     * 是否忽略时间，时间也是公共字段，但值是实时的，需要单独处理
     */
    private boolean ignoreTime = false;

    /**
     * 对缓存列表操作的锁
     */
    private Lock cacheLock = new ReentrantLock(false);

    /**
     * 缓存列表
     */
    private List<String> logCache;

    /**
     * 定时输出缓存日志的定时器
     */
    private ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();

    /**
     * 应用名称：在logback配置中配置
     */
    private String serviceName;

    /**
     * 端口号：在logback配置中配置
     */
    private String port = "8080";

    /**
     * ip地址
     */
    private static String ip;

    static {
        try {
            ip = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            logger.warn("", e);
        }
    }

    /**
     * appender启动及初始化
     * 公共字段拼接、初始化缓存列表、启动定时器
     */
    @Override
    public void start() {
        processCommonFields();

        if (logCache == null) {
            logCache = new ArrayList<>(cacheCapacity);
        }
        scheduledExecutor.scheduleAtFixedRate(() -> {
            cacheLock.lock();
            List<String> cache = logCache;
            logCache = new ArrayList<>(cacheCapacity);
            cacheLock.unlock();
            writeLog(cache);
        }, freq, freq, TimeUnit.SECONDS);
        super.start();
    }

    /**
     * 公共字段拼接字符串处理
     */
    private void processCommonFields() {
        if (!noCommonFields) {
            //固定值的公共字段处理
            Map<String, String> finalFields = new LinkedHashMap<>();
            finalFields.put(SERVICE_NAME, serviceName);
            finalFields.put(IP, ip);
            finalFields.put(PORT, port);

            if (StringUtil.isEmpty(ignoreCommonFields)) {
                genFinalFieldsStr(finalFields);
            } else {
                String[] ignores = ignoreCommonFields.split(",");
                for (String str : ignores) {
                    if (LOG_TIMESTAMP.equals(str)) {
                        ignoreTime = true;
                    }
                    finalFields.remove(str);
                }
                //移除了所有公共字段，即禁止添加公共字段
                if (finalFields.isEmpty()) {
                    noCommonFields = true;
                } else {
                    genFinalFieldsStr(finalFields);
                }
            }
        }
    }

    /**
     * 固定值的公共字段拼接
     *
     * @param finalFields 固定值的公共字段及其值
     */
    private void genFinalFieldsStr(Map<String, String> finalFields) {
        for (Map.Entry<String, String> entry : finalFields.entrySet()) {
            finalFieldsStr = String.format("%s\"%s\":\"%s\",", finalFieldsStr, entry.getKey(), entry.getValue());
        }
    }

    @Override
    protected void subAppend(ILoggingEvent event) {
        if (event == null || StringUtil.isEmpty(event.getFormattedMessage())) {
            return;
        }
        cacheLock.lock();
        //缓存已满需要立即输出，这里使用同步输出
        //只要容量和频率设置合理，缓存溢出情况比较少，使用异步浪费线程资源
        if (logCache.size() > cacheCapacity - 1) {
            //采用新建一个列表的方式快速完成缓存复制和清空
            List<String> cache = logCache;
            logCache = new ArrayList<>(cacheCapacity);
            addCache(event.getFormattedMessage());
            //输出日志为耗时操作，输出前释放锁，防止阻塞
            cacheLock.unlock();
            //此时的cache为局部变量，线程安全
            writeLog(cache);
        } else {
            addCache(event.getFormattedMessage());
            cacheLock.unlock();
        }
    }

    /**
     * 给json日志添加公共字段，然后添加日志到缓存
     *
     * @param msg json日志
     */
    private void addCache(String msg) {
        //禁止添加公共字段，则添加原日志消息
        if (noCommonFields) {
            logCache.add(msg);
        } else {
            msg = msg.trim();
            if (msg.startsWith("{") && msg.endsWith("}")) { //简单判断json格式
                logCache.add(genJsonMsg(msg));
            }
        }
    }

    /**
     * 给日志加上公共字段
     *
     * @param msg 日志
     * @return 加上公共字段后的json日志
     */
    private String genJsonMsg(String msg) {
        if (ignoreTime) {
            return "{" + finalFieldsStr + msg.substring(1);
        } else {
            return "{" + finalFieldsStr + "\"" + LOG_TIMESTAMP + "\":\"" + getCurrentTime() + "\"," + msg.substring(1);
        }
    }

    /**
     * 批量输出缓存列表中的日志到文件
     *
     * @param cache 缓存
     */
    private void writeLog(List<String> cache) {
        if (!cache.isEmpty()) {
            //多条日志用换行符分割拼接成一条
            String allCacheMsg = String.join("\n", cache);
            //定义一个LoggingEvent，日志输出交给父类处理
            LoggingEvent loggingEvent = new LoggingEvent("ch.qos.logback.classic.Logger",
                    logger, Level.INFO, allCacheMsg, null, null);
            super.subAppend(loggingEvent);
        }
    }

    /**
     * 获取当前时间，格式与Filebeat时间格式一致.
     *
     * @return 当前时间
     */
    private String getCurrentTime() {
        DateTimeFormatter dtf = DateTimeFormatter.ofPattern(LOG_TIMESTAMP_FORMAT);
        return dtf.format(LocalDateTime.now());
    }

    public long getFreq() {
        return freq;
    }

    public void setFreq(long freq) {
        this.freq = freq;
    }

    public int getCacheCapacity() {
        return cacheCapacity;
    }

    public void setCacheCapacity(int cacheCapacity) {
        this.cacheCapacity = cacheCapacity;
    }

    public String getServiceName() {
        return serviceName;
    }

    public void setServiceName(String serviceName) {
        this.serviceName = serviceName;
    }

    public String getPort() {
        return port;
    }

    public void setPort(String port) {
        this.port = port;
    }

    public String getIgnoreCommonFields() {
        return ignoreCommonFields;
    }

    public void setIgnoreCommonFields(String ignoreCommonFields) {
        this.ignoreCommonFields = ignoreCommonFields;
    }

    public boolean isNoCommonFields() {
        return noCommonFields;
    }

    public void setNoCommonFields(boolean noCommonFields) {
        this.noCommonFields = noCommonFields;
    }
}
